package com.gzc.just.play.last.war.common.share.communication;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import java.util.Map;
import java.util.UUID;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * 消息总线
 * 负责服务器间的异步通信
 */
@Service
@ConditionalOnClass(RedisConnectionFactory.class)
@ConditionalOnProperty(name = "allinone.enable-redis", havingValue = "true", matchIfMissing = true)
public class MessageBus {

    private static final Logger log = LoggerFactory.getLogger(MessageBus.class);

    private final RedisTemplate<String, Object> redisTemplate;
    private final ObjectMapper objectMapper;
    
    // 线程池
    private final ExecutorService executorService = Executors.newVirtualThreadPerTaskExecutor();

    @Autowired(required = false)
    public MessageBus(RedisTemplate<String, Object> redisTemplate, ObjectMapper objectMapper) {
        this.redisTemplate = redisTemplate;
        this.objectMapper = objectMapper;
    }
    
    // 本地消息队列（替代RabbitMQ）
    private final Map<String, BlockingQueue<Message>> localQueues = new ConcurrentHashMap<>();
    
    // 消息处理器
    private final Map<String, MessageHandler> handlers = new ConcurrentHashMap<>();
    
    // Redis键前缀
    private static final String MESSAGE_BUS_KEY = "message_bus:";
    private static final String RESPONSE_KEY = "response:";
    
    /**
     * 初始化
     */
    public void init() {
        // 启动消息处理线程
        executorService.submit(this::processLocalMessages);
        
        log.info("MessageBus initialized");
    }
    
    /**
     * 发送消息（异步）
     */
    public CompletableFuture<Void> sendMessageAsync(String targetService, String messageType, Object message) {
        return CompletableFuture.runAsync(() -> {
            try {
                sendMessage(targetService, messageType, message);
            } catch (Exception e) {
                log.error("Error sending async message", e);
                throw new RuntimeException(e);
            }
        }, executorService);
    }
    
    /**
     * 发送消息（同步）
     */
    public void sendMessage(String targetService, String messageType, Object message) {
        try {
            // 创建消息对象
            Message msg = new Message();
            msg.id = UUID.randomUUID().toString();
            msg.type = messageType;
            msg.targetService = targetService;
            msg.payload = message;
            msg.timestamp = System.currentTimeMillis();
            
            // 序列化消息
            String messageJson = objectMapper.writeValueAsString(msg);
            
            // 如果是本地服务，直接加入本地队列
            if (isLocalService(targetService)) {
                String queueName = MESSAGE_BUS_KEY + targetService;
                BlockingQueue<Message> queue = localQueues.computeIfAbsent(queueName, k -> new LinkedBlockingQueue<>());
                queue.offer(msg);
                log.debug("Added message to local queue: {}", queueName);
            } else {
                // 远程服务，使用Redis发布
                String channel = MESSAGE_BUS_KEY + targetService;
                redisTemplate.convertAndSend(channel, messageJson);
                log.debug("Published message to Redis channel: {}", channel);
            }
            
            log.debug("Sent message: {} to service: {}", messageType, targetService);
        } catch (JsonProcessingException e) {
            log.error("Error serializing message", e);
            throw new RuntimeException("Error serializing message", e);
        }
    }
    
    /**
     * 发送请求并等待响应
     */
    public <T> CompletableFuture<T> sendRequest(String targetService, String messageType, Object message, Class<T> responseType) {
        return CompletableFuture.supplyAsync(() -> {
            try {
                // 创建请求消息
                Message request = new Message();
                request.id = UUID.randomUUID().toString();
                request.type = messageType;
                request.targetService = targetService;
                request.payload = message;
                request.timestamp = System.currentTimeMillis();
                request.request = true;
                
                // 设置响应回调
                String responseKey = RESPONSE_KEY + request.id;
                
                // 注册响应处理器
                CompletableFuture<T> responseFuture = new CompletableFuture<>();
                handlers.put(responseKey, new MessageHandler() {
                    @Override
                    public void handle(Message response) {
                        try {
                            T responseData = objectMapper.convertValue(response.payload, responseType);
                            responseFuture.complete(responseData);
                        } catch (Exception e) {
                            responseFuture.completeExceptionally(e);
                        }
                    }
                });
                
                // 发送请求
                sendMessage(targetService, messageType, request);
                
                // 等待响应（带超时）
                return responseFuture.get(30, TimeUnit.SECONDS);
            } catch (Exception e) {
                log.error("Error sending request", e);
                throw new RuntimeException("Error sending request", e);
            }
        }, executorService);
    }
    
    /**
     * 注册消息处理器
     */
    public void registerHandler(String messageType, MessageHandler handler) {
        handlers.put(messageType, handler);
        log.info("Registered handler for message type: {}", messageType);
    }
    
    /**
     * 注销消息处理器
     */
    public void unregisterHandler(String messageType) {
        handlers.remove(messageType);
        log.info("Unregistered handler for message type: {}", messageType);
    }
    
    /**
     * 处理本地消息
     */
    private void processLocalMessages() {
        while (!Thread.currentThread().isInterrupted()) {
            try {
                for (Map.Entry<String, BlockingQueue<Message>> entry : localQueues.entrySet()) {
                    String queueName = entry.getKey();
                    BlockingQueue<Message> queue = entry.getValue();
                    
                    Message message = queue.poll(1, TimeUnit.SECONDS);
                    if (message != null) {
                        processMessage(message);
                    }
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            } catch (Exception e) {
                log.error("Error processing local messages", e);
            }
        }
    }
    
    /**
     * 处理消息
     */
    private void processMessage(Message message) {
        try {
            String messageType = message.type;
            
            // 处理响应消息
            if (message.responseTo != null) {
                String responseKey = RESPONSE_KEY + message.responseTo;
                MessageHandler handler = handlers.get(responseKey);
                if (handler != null) {
                    handler.handle(message);
                    handlers.remove(responseKey); // 移除一次性处理器
                }
                return;
            }
            
            // 处理普通消息
            MessageHandler handler = handlers.get(messageType);
            if (handler != null) {
                handler.handle(message);
            } else {
                log.warn("No handler found for message type: {}", messageType);
            }
        } catch (Exception e) {
            log.error("Error processing message", e);
        }
    }
    
    /**
     * 发送响应
     */
    public void sendResponse(Message originalMessage, Object response) {
        Message responseMsg = new Message();
        responseMsg.id = UUID.randomUUID().toString();
        responseMsg.type = originalMessage.type + "_RESPONSE";
        responseMsg.targetService = originalMessage.sourceService;
        responseMsg.payload = response;
        responseMsg.responseTo = originalMessage.id;
        
        // 发送响应
        sendMessage(originalMessage.sourceService, responseMsg.type, responseMsg);
    }
    
    /**
     * 判断是否是本地服务
     */
    private boolean isLocalService(String serviceName) {
        // 这里应该根据实际的服务名判断
        // 简单实现：如果包含"local"或者某些特定后缀，认为是本地服务
        return serviceName.contains("local") || serviceName.endsWith("-1");
    }
    
    /**
     * 获取统计信息
     */
    public Map<String, Object> getStats() {
        Map<String, Object> stats = new ConcurrentHashMap<>();
        
        // 队列大小
        Map<String, Integer> queueSizes = new ConcurrentHashMap<>();
        for (Map.Entry<String, BlockingQueue<Message>> entry : localQueues.entrySet()) {
            queueSizes.put(entry.getKey(), entry.getValue().size());
        }
        stats.put("queueSizes", queueSizes);
        
        // 处理器数量
        stats.put("handlerCount", handlers.size());
        
        return stats;
    }
    
    /**
     * 关闭消息总线
     */
    public void shutdown() {
        executorService.shutdown();
        try {
            if (!executorService.awaitTermination(10, TimeUnit.SECONDS)) {
                executorService.shutdownNow();
            }
        } catch (InterruptedException e) {
            executorService.shutdownNow();
            Thread.currentThread().interrupt();
        }
        
        log.info("MessageBus shutdown");
    }
}